package com.niit.service

import com.niit.bean.AnswerWithRecommendations
import com.niit.dao.EDUBatchDao
import com.niit.utils.SparkUtil
import org.apache.spark.sql.{DataFrame, Dataset, Row}

class EDUBatchService {

  private val spark = SparkUtil.takeSpark()
  import spark.implicits._
  import org.apache.spark.sql.functions._

  def DataAnalysis(): Unit ={
    val eduDao = new EDUBatchDao
    val answerData: Dataset[AnswerWithRecommendations] = eduDao.getEDUData()
    hotSubjectCountTop50(answerData)
    hostSubjectRecommendTop20(answerData)
  }

  /*
   需求一：找到Top50热点题对应的科目。
   1. 先找到 出现次数最后的前50个 题目
   2.根据 前50的题目 找到 对应 科目
   3. 找到热点题 对应科目 出现的 次数
   */
  def hotSubjectCountTop50(answerData: Dataset[AnswerWithRecommendations]): Unit ={
    //1. 找到 出现次数最多的前50个 题目
    val hotTop50: Dataset[Row] = answerData.groupBy("question_id")
      .agg(count("*") as "hot")
      .orderBy('hot.desc).limit(50)
    //结果 question_id   count
    //2.根据 前50的题目 找到 对应 科目
    val joinDF: DataFrame = hotTop50.join(answerData.dropDuplicates("question_id"), "question_id")
    //结果 student_id grade_id subject_id textboot_id  question_id  count
  // 3. 找到热点题 对应科目 出现的 次数
    val res: Dataset[Row] = joinDF.groupBy("subject_id").agg(count("*") as "hotCount").orderBy('hotCount.desc)
    println("需求一：找到Top50热点题对应的科目")
    res.show()

  }


  /*
    需求二：各个科目推荐分析
     找的前20热点题对应的推荐题目，然后找到推荐题目对应的科目，并统计每个科目分别包含热点题目的数量
   */
  def hostSubjectRecommendTop20(answerData: Dataset[AnswerWithRecommendations]): Unit ={
    //1. 找到前20的热点题
    val hotTop20: Dataset[Row] = answerData.groupBy("question_id").agg(count("*") as "hot").orderBy('hot.desc).limit(20)

    //结果： question_id   count
    //2.根据前20的question_id  找到对应的推荐列表
    val ridsDF: DataFrame = hotTop20.join(answerData, "question_id").select("recommendations")

    //结果：recommendations  格式   ：题目ID_2,题目ID_4,题目ID_1,题目ID_8,题目ID_7,…………
    //3.根据 ridsDF 进行以 , 切割字符串（推荐列表） 目的根据题目 去找到 对应 科目
    val ridsDS: Dataset[Row] = ridsDF.select(explode(split('recommendations, ",") )  as "question_id").dropDuplicates("question_id")
    //结果:将推荐列表以逗号切割后的 question_id
    val ridAndSid: DataFrame = ridsDS.join(answerData.dropDuplicates("question_id"), "question_id")
    //结果 quesion_id  subject_id student_id grade_id……………………
    //4.统计科目出现的次数
    val res: Dataset[Row] = ridAndSid.groupBy("subject_id").agg(count("*") as "rcount").orderBy('rcount.desc)
    println("需求二：各个科目推荐分析")
    res.show()
  }


}
